package com.teamsun.mqtt;

import java.net.InetSocketAddress;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.springframework.beans.factory.annotation.Autowired;

import com.teamsun.mqtt.handle.ConnectionService;
import com.teamsun.mqtt.handle.PushServiceHandle;
import com.teamsun.mqtt.handle.ReadAndWriteTimeoutHandle;
import com.teamsun.mqtt.handle.SubServiceHandle;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.mqtt.MqttDecoder;
import io.netty.handler.codec.mqtt.MqttEncoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;

/**
 * 
 * @author tzj
 * 
 */
public class MqttServer {


	/**
	 * 数据库操作线程
	 */
	final ExecutorService dboptGroup = Executors.newFixedThreadPool(8);
	
	final ConcurrentNavigableMap<String, BlockingQueue<String>> submap = new ConcurrentSkipListMap<String, BlockingQueue<String>>();
	
	/**
	 * 连接请求处理的handle
	 */
	@Autowired
	ConnectionService connectionService;
	
	/**
	 * 订阅请求处理的handle
	 */
	@Autowired
	SubServiceHandle subServiceHandle;
	
	/**
	 * 推送消息处理的handle
	 */
	@Autowired
	PushServiceHandle serviceHandle;
	
	
	private int sobacklog;

	private int port;

	private int bossthreads;

	private int dbthreads;

	private int workthreads;

	private int readtimeout;

	public void start(int port) throws InterruptedException {

		ServerBootstrap bootstrap = new ServerBootstrap();// 启动辅助类

		// io线程和工作线程
		NioEventLoopGroup group = new NioEventLoopGroup(bossthreads);

	
		final NioEventLoopGroup workGroup = new NioEventLoopGroup(workthreads);


		bootstrap.group(group, workGroup);
		bootstrap.channel(NioServerSocketChannel.class);//
		bootstrap.localAddress(new InetSocketAddress(port));// 绑定端口
		bootstrap.option(ChannelOption.SO_BACKLOG, sobacklog);

		bootstrap.childOption(ChannelOption.SINGLE_EVENTEXECUTOR_PER_GROUP,
				false);
		try {
			bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {//
						protected void initChannel(SocketChannel ch)
								throws Exception {
							
							ch.pipeline()
									.addLast(new ReadAndWriteTimeoutHandle(readtimeout))
									.addLast(workGroup,MqttEncoder.INSTANCE,
											new MqttDecoder(),
											connectionService,
											subServiceHandle, serviceHandle);

						}

					});
			ChannelFuture f = bootstrap.bind().sync();// 实际绑定操作
			System.out.println(MqttServer.class.getName() + " 服务器启动成功 "
					+ f.channel().localAddress());

			f.channel().closeFuture().sync();// ;// 等待 服务器关闭
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			group.shutdownGracefully().sync();// 优雅关闭
			workGroup.shutdownGracefully().sync();
		}
	}

	public int getSobacklog() {
		return sobacklog;
	}

	public void setSobacklog(int sobacklog) {
		this.sobacklog = sobacklog;
	}

	public int getPort() {
		return port;
	}

	public void setPort(int port) {
		this.port = port;
	}

	public int getBossthreads() {
		return bossthreads;
	}

	public void setBossthreads(int bossthreads) {
		this.bossthreads = bossthreads;
	}

	public int getDbthreads() {
		return dbthreads;
	}

	public void setDbthreads(int dbthreads) {
		this.dbthreads = dbthreads;
	}

	public int getWorkthreads() {
		return workthreads;
	}

	public void setWorkthreads(int workthreads) {
		this.workthreads = workthreads;
	}

	public int getReadtimeout() {
		return readtimeout;
	}

	public void setReadtimeout(int readtimeout) {
		this.readtimeout = readtimeout;
	}

	public ExecutorService getDboptGroup() {
		return dboptGroup;
	}

	public ConcurrentNavigableMap<String, BlockingQueue<String>> getSubmap() {
		return submap;
	}

	public ConnectionService getConnectionService() {
		return connectionService;
	}

	public void setConnectionService(ConnectionService connectionService) {
		this.connectionService = connectionService;
	}

	public SubServiceHandle getSubServiceHandle() {
		return subServiceHandle;
	}

	public void setSubServiceHandle(SubServiceHandle subServiceHandle) {
		this.subServiceHandle = subServiceHandle;
	}

	public PushServiceHandle getServiceHandle() {
		return serviceHandle;
	}

	public void setServiceHandle(PushServiceHandle serviceHandle) {
		this.serviceHandle = serviceHandle;
	}

	
	
}